package com.gugee.ins.data.analysis.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.AiConst;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.ai.AiAnalysisClient;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.comment.InsCommentDao;
import com.gugee.ins.common.dal.service.media.InsMediaDao;
import com.gugee.ins.common.model.ai.InsBloggerLanguage;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerMain;
import com.gugee.ins.common.model.mysql.comment.InsComment;
import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.InsAnalysisLanguageService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBloggerLocation 服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsAnalysisLanguageServiceImpl implements InsAnalysisLanguageService {

    /**
     * 配置
     */
    @Autowired
    InsAnalysisConfig insAnalysisConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * ai
     */
    @Autowired
    AiAnalysisClient aiAnalysisClient;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     * InsMediaDao
     */
    @Autowired
    InsMediaDao insMediaDao;

    /**
     * insCommentDao
     */
    @Autowired
    InsCommentDao insCommentDao;
    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsAnalysisAvatar;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insAnalysisLanguage-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insAnalysisConfig.getInsAnalysisLanguageCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理 分析语言 任务
     * @param insBloggerMessages
     */
    @Override
    public void handleAnalysisLanguage(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    // 分析语言
                    analysisLanguage(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("insAnalysisLanguage error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("insAnalysisLanguage thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 分析语言
     *
     * @param insBloggerMessage
     */
    public void analysisLanguage(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();

        // 判断是否需要处理
        if(insBloggerMessage.getTypeCode() != InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode()){
            // 查询博主帖子，将帖子的title内容（文字长度在一定范围的）发送给ai接口分析语言
            analysisMediaLanguage(uid);
        }

        // 将博主活跃粉丝的评论发送给ai接口分析语言
        analysisCommentLanguage(uid);

        // 优先统计语言
        sendPriorityStMsg(insBloggerMessage);

        if ((System.currentTimeMillis() - start) / 1000 > insAnalysisConfig.getInsAnalysisLanguageDebugTimeLimit()) {
            log.info("insAnalysisLanguage uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 分析博主帖子标题上的语言
     * @param uid
     */
    public void analysisMediaLanguage(long uid){
        // 判断是否分析过，分析过就不再分析
        InsBloggerMain insBlogger = insBloggerMainDao.findByUid(uid);
        if(insBlogger != null && (StringUtils.isEmpty(insBlogger.getLanguage())
                        || AiConst.LANGUAGE_UNKNOWN.equals(insBlogger.getLanguage()))){
            // 获取帖子标题
            List<String> captions = insMediaDao.findCaptionByUid(uid,insAnalysisConfig.getInsAnalysisLanguageTextMinLength());
            if(captions != null && !captions.isEmpty()){
                Map<String,String> aiAnalysisMap = new HashMap<>(captions.size());
                String caption;
                // 拼装语言分析请求参数
                for(int i = 0; i < captions.size(); ++i){
                    caption = captions.get(i);
                    aiAnalysisMap.put(uid + "_" + i,caption.length() > 50 ? caption.substring(0,50) : caption);
                    if(aiAnalysisMap.size() > insAnalysisConfig.getInsAnalysisLanguageTextCount()){
                        break;
                    }
                }
                if(!aiAnalysisMap.isEmpty()){
                    try {
                        // 调用ai接口
                        Map<String,String> result = aiAnalysisClient.getLanguageByText(aiAnalysisMap);
                        String bloggerLanguage = AiConst.LANGUAGE_UNKNOWN;
                        if(!result.isEmpty()){
                            Map<String,Integer> languageMap = new HashMap<>(result.size());
                            String language;
                            Iterator<String> iterator = result.values().iterator();
                            int max = 0;
                            // 选取最多的那个语言
                            while(iterator.hasNext()){
                                language = iterator.next();
                                if(!StringUtils.isEmpty(language)){
                                    int count = languageMap.getOrDefault(language,0);
                                    languageMap.put(language,count + 1);
                                    if(count == max){
                                        max++;
                                        bloggerLanguage = AiConst.LANGUAGE_UNKNOWN.equals(language) ? bloggerLanguage : language;
                                    }
                                }
                            }
                        }
                        // 更新保存语言
                        updateLanguage(uid,bloggerLanguage);
                    }catch (Exception e){
                        log.error("insAnalysisLanguage get media caption ai result error, msg:{}",e.getMessage());
                    }
                }
            }
        }
    }

    /**
     * 分析博主活跃粉丝评论的语言
     * @param uid
     */
    public void analysisCommentLanguage(long uid){
        // 获取未分析出语言的活跃粉丝的id及其的评论
        List<InsComment> insComments = insCommentDao.findAnalysisLanguageCommentByUid(uid);
        if(insComments != null && !insComments.isEmpty()){
            long cUid;
            Set<Long> cUids = new HashSet<>(insComments.size());
            Map<String,String> aiAnalysisMap = new HashMap<>(insAnalysisConfig.getInsAnalysisLanguageTextCount());

            // 循环批量调用ai分析语言接口
            for(int i = 0; i < insComments.size(); ++i){
                cUid = insComments.get(i).getCUid();
                if(!cUids.contains(cUid)){
                    cUids.add(cUid);
                    aiAnalysisMap.put(cUid + "",insComments.get(i).getText());
                    if(aiAnalysisMap.size() == insAnalysisConfig.getInsAnalysisLanguageTextCount()){
                        analysisLanguage(aiAnalysisMap);
                        aiAnalysisMap = new HashMap<>(insAnalysisConfig.getInsAnalysisLanguageTextCount());
                    }
                }
            }
            if(!aiAnalysisMap.isEmpty()){
                analysisLanguage(aiAnalysisMap);
            }
        }
    }

    /**
     * 分批次分析评论的语言
     * @param aiAnalysisMap
     */
    public void analysisLanguage(Map<String,String> aiAnalysisMap){
        try{
            // 调用ai接口
            Map<String,String> result = aiAnalysisClient.getLanguageByText(aiAnalysisMap);
            if(!result.isEmpty()){
                Iterator<Map.Entry<String,String>> iterator = result.entrySet().iterator();
                List<InsBloggerLanguage> insBloggerLanguages = new ArrayList<>(result.size());
                while(iterator.hasNext()){
                    try{
                        Map.Entry<String,String> entry = iterator.next();
                        InsBloggerLanguage insBloggerLanguage = new InsBloggerLanguage();
                        insBloggerLanguage.setLanguage(entry.getValue());
                        insBloggerLanguage.setUid(Long.parseLong(entry.getKey()));
                        insBloggerLanguages.add(insBloggerLanguage);
                    }catch (Exception e){
                        log.error("ai analysis language result uid error, msg:{}",e.getMessage());
                    }
                }

                // 更新保存语言
                if(!insBloggerLanguages.isEmpty()){
                    updateActiveFansLanguage(insBloggerLanguages);
                }
            }
        }catch (Exception ex){
            log.error("insAnalysisLanguage get comment text ai result error, msg:{}",ex.getMessage());
        }
    }

    /**
     * 更新语言
     * @param uid
     * @param language
     */
    public void updateLanguage(long uid,String language){
        try{
            //更新数据库 blogger
            insBloggerDao.updateLanguage(uid,language);
            //更新数据库 blogger_main
            insBloggerMainDao.updateLanguage(uid,language);
        }catch (Exception e){
            log.error("update insAnalysisLanguage language error, msg:{} ",e.getMessage());
        }
    }

    /**
     * 批量更新语言
     * @param insBloggerLanguages
     */
    public void updateActiveFansLanguage(List<InsBloggerLanguage> insBloggerLanguages){
        try{
            //更新数据库 blogger
            insBloggerDao.batchUpdateLanguage(insBloggerLanguages);
            //更新数据库 blogger_main
            insBloggerMainDao.batchUpdateLanguage(insBloggerLanguages);
        }catch (Exception e){
            log.error("batch update insAnalysisLanguage language error, msg:{} ",e.getMessage());
        }
    }

    /**
     * 发送优先统计语言的消息
     * @param insBloggerMessage
     */
    public void sendPriorityStMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.ST_LANGUAGE.getCode());
        insBloggerMessage.setRetryTimes(insAnalysisConfig.getInsMBloggerActiveFollowerDataStPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insAnalysisConfig.getInsMBloggerActiveFollowerDataStPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsAnalysisAvatar.send(insAnalysisConfig.getInsMBloggerActiveFollowerDataStPriorityTopic(),insBloggerMessage);
    }
}
